package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.VideoLookProcessBean;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.ZoneId;

/**
 * @author zzw
 * @create 2022/9/4 10:35
 */
public class DwsVideoLookProcessWindow {
    public static void main(String[] args) {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        env.setStateBackend(new HashMapStateBackend());
        */

        // 设置时区的方法
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 3 读取DWD层 dwd_video_look_process 主题数据
        String topicName = "dwd_video_look_process";
        String groupId = "dws_video_look_process_window";
        DataStreamSource<String> lookProcessStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        // TODO 4 转换数据类型
        SingleOutputStreamOperator<JSONObject> jsonObjStream = lookProcessStream.map(JSON::parseObject);

        // TODO 5 根据mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(JSONObject -> JSONObject.getJSONObject("common").getString("uid"));

        SingleOutputStreamOperator<VideoLookProcessBean> processStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, VideoLookProcessBean>() {
            // ValueState<JSONObject> lastLookDtState = null;
            ValueState<JSONObject> lastLookUidState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                // lastLookDtState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("last_look_dt", JSONObject.class));
                lastLookUidState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("last_look_uid", JSONObject.class));
            }

            @Override
            public void processElement(JSONObject value, Context ctx, Collector<VideoLookProcessBean> out) throws Exception {
                //JSONObject lastLookDt = lastLookDtState.value();
                String lastLookUid = lastLookUidState.value().toJSONString();
                Long playSec = value.getJSONObject("appVideo").getLong("play_sec");
                String uid = value.getJSONObject("common").getString("uid");
                String videoId = value.getJSONObject("appVideo").getString("vide_id");
                Long ts = value.getLong("ts");
                long usCount = 0;
                if (playSec != 30) {
                    //是一条播放数据
                    if (uid.equals(lastLookUid)) {
                        usCount = 1L;
                        out.collect(VideoLookProcessBean.builder()
                                .videoLookCount(1L)
                                .totalTimeCount(playSec)
                                .usCount(1L)
                                .avgTime(0L)
                                .build());
                    } else {
                        out.collect(VideoLookProcessBean.builder()
                                .videoLookCount(1L)
                                .totalTimeCount(playSec)
                                .usCount(0L)
                                .avgTime(0L)
                                .build());
                    }
                } else {
                    //后续数据进入输出观看时长
                    out.collect(VideoLookProcessBean.builder()
                            .videoLookCount(1L)
                            .totalTimeCount(playSec)
                            .usCount(1L)
                            .avgTime(0L)
                            .build());
                }
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<VideoLookProcessBean> out) throws Exception {
                lastLookUidState.clear();
            }
        });

        // 开窗聚合
        SingleOutputStreamOperator<VideoLookProcessBean> beanwithWaterMark = processStream.assignTimestampsAndWatermarks(WatermarkStrategy.<VideoLookProcessBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<VideoLookProcessBean>() {
            @Override
            public long extractTimestamp(VideoLookProcessBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        beanwithWaterMark.windowAll();
    }
}
